package vender

import (
	"errors"
	"fmt"
	"sync"

	"github.com/streadway/amqp"
)

type Rabbitmq struct {
	Address    string // 链接
	QueueName  string // 队列名
	MaxRetry   int    // 最大重试
	MaxConsume int    // 最大消费者
}

var (
	rabbitmqConn  = make(map[string]*amqp.Connection)
	rabbitmqCh    = make(map[string][]*amqp.Channel)
	rabbitmqQueue = make(map[string]*amqp.Queue)
)

// 初始化队列
func (r *Rabbitmq) InitRabbitmq() error {
	// 拨号
	conn, err := amqp.Dial(r.Address)
	if err != nil {
		return err
	}
	rabbitmqConn[r.QueueName] = conn
	// 信道, 只创建1个信道，如果配置多个消费者，在启动消费者的时候，再增加
	ch, err := conn.Channel()
	if err != nil {
		return err
	}
	rabbitmqCh[r.QueueName] = append(rabbitmqCh[r.QueueName], ch)
	return nil
}

// 关闭连接
func (r *Rabbitmq) Close() {
	conn, ok := rabbitmqConn[r.QueueName]
	if ok {
		conn.Close()
	}
	chLs, ok := rabbitmqCh[r.QueueName]
	if ok {
		for _, ch := range chLs {
			ch.Close()
		}
	}
}

// 获取默认生产队列
func (r *Rabbitmq) DefaultQueue() error {
	ch := rabbitmqCh[r.QueueName][0]
	// 验证队列是否ok
	q, err := ch.QueueDeclare(
		r.QueueName, // 队列名字
		true,        // 消息是否持久化
		false,       // 不使用的时候删除队列
		false,       // exclusive
		false,       // no-wait
		nil,         // arguments
	)
	if err != nil {
		return err
	}
	rabbitmqQueue[r.QueueName] = &q
	return nil
}

// 默认生产者发消息
func (r *Rabbitmq) DefaultProducer(contentType string, body []byte) error {
	q, ok := rabbitmqQueue[r.QueueName]
	if !ok {
		return errors.New(r.QueueName + "默认队列不存在")
	}
	chLs, ok := rabbitmqCh[r.QueueName]
	if !ok {
		return errors.New(r.QueueName + "默认信道不存在")
	}
	ch := chLs[0]
	err := ch.Publish(
		"",     // exchange（交换机名字），这里忽略
		q.Name, // 路由参数，这里使用队列名字作为路由参数
		false,  // mandatory 强制
		false,  // immediate 实时
		amqp.Publishing{
			Headers: amqp.Table{
				"retry_count": r.MaxRetry, // 重试次数
			},
			DeliveryMode: 2,
			ContentType:  contentType,
			Body:         body, // 消息内容
		},
	)
	if err != nil {
		return err
	}
	return nil
}

// 默认消费者
func (r *Rabbitmq) DefaultConsumer(hander func(info []byte) error) error {
	q, ok := rabbitmqQueue[r.QueueName]
	if !ok {
		return errors.New(r.QueueName + "默认队列不存在")
	}
	_, ok = rabbitmqCh[r.QueueName]
	if !ok {
		return errors.New(r.QueueName + "默认信道不存在")
	}
	// 如果消费者配置了多个，创建多个信道
	for i := len(rabbitmqCh[r.QueueName]); i < r.MaxConsume; i++ {
		conn, ok := rabbitmqConn[r.QueueName]
		if !ok {
			return errors.New(r.QueueName + "链接不存在")
		}
		ch, err := conn.Channel()
		if err != nil {
			return err
		}
		rabbitmqCh[r.QueueName] = append(rabbitmqCh[r.QueueName], ch)
	}
	chLs := rabbitmqCh[r.QueueName]
	var wg sync.WaitGroup
	for i, ch := range chLs {
		wg.Add(1)
		go func(ch *amqp.Channel, i int) {
			defer wg.Done()
			msgs, err := ch.Consume(
				q.Name,                                  // 队列名
				fmt.Sprintf("%v_consumer%v", q.Name, i), // 消费者名字，不填，则自动生成一个唯一ID
				false,                                   // 是否自动提交消息，即自动告诉rabbitmq消息已经处理成功。
				false,                                   // exclusive
				false,                                   // no-local
				false,                                   // no-wait
				nil,                                     // args
			)
			if err != nil {
				panic(err)
			}
			for msg := range msgs {
				info := fmt.Sprintf("消费者:%v, 消息为:%s", msg.ConsumerTag, msg.Body)
				if err := hander([]byte(info)); err != nil {
					r.consumeErr(ch, q, msg)
				} else {
					msg.Ack(false)
				}
			}
		}(ch, i)
	}
	wg.Wait()
	return nil
}

// 处理消费失败情况
// 思路：消费失败，如果有重试则重新入队，如果没有则直接丢弃，并发布消息
func (r *Rabbitmq) consumeErr(ch *amqp.Channel, q *amqp.Queue, msg amqp.Delivery) {
	if msg.Headers["retry_count"].(int32) <= 0 {
		fmt.Println("消息重试失败最大次数, 进入失败缓存" + string(msg.Body))
		msg.Reject(false)
		return
	}
	err := ch.Publish(
		"",     // exchange（交换机名字），这里忽略
		q.Name, // 路由参数，这里使用队列名字作为路由参数
		false,  // mandatory 强制
		false,  // immediate 实时
		amqp.Publishing{
			Headers: amqp.Table{
				"retry_count": msg.Headers["retry_count"].(int32) - 1, // 重试次数
			},
			DeliveryMode: 2,
			ContentType:  msg.ContentType,
			Body:         msg.Body, // 消息内容
		},
	)
	if err != nil {
		fmt.Println("消息重试失败, 进入失败缓存" + string(msg.Body) + " " + err.Error())
		msg.Reject(false)
	}
	msg.Ack(false)
}
